import java.util.Map ;
import java.util.HashMap;
import java.util.Properties ;
import java.util.Arrays ;
import java.util.List ;
import java.util.ArrayList ;

import java.time.Duration ;

import io.github.cdimascio.dotenv.Dotenv;
import lib.avro.Avro;
import lib.common.Common;

import proto.online.OnlineGrpc ;
import proto.push.PushOuterClass.AvroBulkCreate;
import proto.push.PushOuterClass.JsonBulkCreate;

import proto.xdefault.Xdefault.DefaultReply;

import io.grpc.stub.StreamObserver;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericContainer ;

import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.kafka.clients.producer.KafkaProducer ;
import org.apache.kafka.clients.producer.ProducerRecord ;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.clients.consumer.ConsumerRecord ;

import org.codehaus.janino.ExpressionEvaluator;

import org.apache.avro.generic.GenericDatumWriter;

import java.util.Set ;
import java.util.HashSet ;

import core.Combinator;
import core.Combinator.StateHandler ;
import core.Combinator.WholeAnd;
import core.Combinator.RollAnd ;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.BlockingQueue;

import com.linecorp.armeria.server.Server ;
import com.linecorp.armeria.common.HttpResponse ;
import com.linecorp.armeria.server.grpc.GrpcService;
import com.linecorp.armeria.common.HttpStatus;

import org.apache.logging.log4j.Logger ;
import org.apache.logging.log4j.LogManager ;


public class App {
    private static final String MODULE = "online-server-java" ;
    private static final Logger LOGGER = LogManager.getLogger(LogManager.ROOT_LOGGER_NAME) ;
    private static void info(String message) {
        LOGGER.info("[component:{}] {}", MODULE, message) ;
    } 
    private static void warn(String message) {
        LOGGER.warn("[component:{}] {}", MODULE, message) ;
    }
    private static void error(String message) {
        LOGGER.error("[component:{}] {}", MODULE, message) ;
    }
    public static String extractPlatformName(String schemaName) {
        String result = "unknown" ;
        if (schemaName.equals("DcSdkAndroid")) {
            result = "android" ;
        } else if (schemaName.equals("DcSdkIos")) {
            result = "ios" ;
        } else if (schemaName.equals("DcSdkWeb")) {
            result = "web" ;
        } else if (schemaName.equals("DcSdkWechat")) {
            result = "wechat" ;
        } else if (schemaName.equals("DcSdkH5")) {
            result = "h5" ;
        }
        return result ;
    }

    public Map<Object, Object> rules = new HashMap<Object, Object>() ;

    public static StateHandler mkHandler(Object ruleVal) throws Exception {
        // WholeAnd handler = new WholeAnd() ;
        RollAnd handler = new RollAnd() ;
        info("ruleVal:" + ruleVal) ;
        Map<String, Object> ruleValInfo = (Map<String, Object>) ruleVal ;
        String appId = ruleValInfo.get("appId").toString() ;
        for(Object rowRule: (List<Object>) ruleValInfo.get("eventRuleList")) {
            Set<String> paramNames = new HashSet<String>() ;
            List<Class> paramTypes = new ArrayList<Class>() ;
            paramNames.add("app_id") ;
            paramTypes.add(String.class) ;
            paramNames.add("event_code") ;
            paramTypes.add(String.class) ;
            
            ExpressionEvaluator ee = new ExpressionEvaluator();
            ee.setExpressionType(boolean.class) ;

            Map<String, Object> rowRuleInfo = (Map<String, Object>) rowRule ;
            String rowExpr = null ;
            String eventCode = rowRuleInfo.get("eventCode").toString() ;
            for(Object rule: (List<Object>) rowRuleInfo.get("rule")) {
                Map<String, String> ruleInfo = (Map<String, String>) rule ;
            
                String ruleAttrName = ruleInfo.get("attribute") ;
                String ruleAttrText = ruleInfo.get("attributeValue") ;
                String ruleAttrType = ruleInfo.get("attributeType") ;
                
                String ruleOperator = ruleInfo.get("operator") ;
                
                String ruleAttrVal = "_delay_" ;
                
                if (!paramNames.contains(ruleAttrName) ) {
                    paramNames.add(ruleAttrName) ;
                    if (ruleAttrType.equals("string")) {
                        ruleAttrVal = "\"" + ruleAttrText + "\"" ;
                        paramTypes.add(String.class) ;
                    } else  if (ruleAttrType.equals("long")) {
                        ruleAttrVal = ruleAttrText ;
                        paramTypes.add(long.class) ;
                    } else if (ruleAttrType.equals("double")) {
                        ruleAttrVal = ruleAttrText ;
                        paramTypes.add(double.class) ;
                    }
                }
                String currentExpr = String.format("%s %s %s", ruleAttrName, ruleOperator, ruleAttrVal) ;
                if (ruleOperator.equals("0")) {
                    currentExpr = String.format("%s.compareTo(%s) < 0", ruleAttrName, ruleAttrVal) ;
                } else if (ruleOperator.equals("1")) {
                    currentExpr = String.format("%s.compareTo(%s) <= 0", ruleAttrName, ruleAttrVal) ;
                } else if (ruleOperator.equals("2")) {
                    currentExpr = String.format("%s.compareTo(%s) == 0", ruleAttrName, ruleAttrVal) ;
                } else if (ruleOperator.equals("3")) {
                    currentExpr = String.format("%s.compareTo(%s) > 0", ruleAttrName, ruleAttrVal) ;
                } else if (ruleOperator.equals("4")) {
                    currentExpr = String.format("%s.compareTo(%s) >= 0", ruleAttrName, ruleAttrVal) ;
                } else if (ruleOperator.equals("5")) {
                    currentExpr = String.format("%s.compareTo(%s) != 0", ruleAttrName, ruleAttrVal) ;
                } else if (ruleOperator.equals("6")) {
                    currentExpr = String.format("%s.contains(%s)", ruleAttrName, ruleAttrVal) ;
                } else if (ruleOperator.equals("7")) {
                    currentExpr = String.format("! %s.contains(%s)", ruleAttrName, ruleAttrVal) ;
                } else if (ruleOperator.equals("8")) {
                    currentExpr = String.format("%s != null && ! %s.isEmpty()", ruleAttrName, ruleAttrName) ;
                } else if (ruleOperator.equals("9")) {
                    currentExpr = String.format("%s != null && %s.isEmpty()", ruleAttrName, ruleAttrName) ;
                } else if (ruleOperator.equals("10")) {
                    currentExpr = String.format("%s == null", ruleAttrName) ;
                } else if (ruleOperator.equals("11")) {
                    currentExpr = String.format("%s != null", ruleAttrName) ;
                }

                rowExpr = (rowExpr == null)?  currentExpr : String.format("%s && (%s)", rowExpr, currentExpr) ;
            }
            if (rowExpr == null) {
                rowExpr = String.format("(app_id.equals(%s)) && (event_code.equals(%s))", "\"" + appId + "\"", "\"" + eventCode + "\"") ;
            } else {
                rowExpr = String.format("(app_id.equals(%s)) && (event_code.equals(%s)) && (%s)", "\"" + appId + "\"", "\"" + eventCode + "\"", rowExpr ) ;
            }

            ee.setParameters(paramNames.toArray(new String[0]), paramTypes.toArray(new Class[0]));

            info("rowExpr:" + rowExpr) ;
            ee.cook(rowExpr) ;
            
            handler.addItem(ee, paramNames) ;
        }

        return handler ;
    }

    public void asyncLoadRules(final String kafkas, String topic) throws Exception {
        info("wait 5 seconds to load...") ;
        new Thread(new Runnable() {
            @Override
            public void run()  {
                Properties props = new Properties() ;
                props.setProperty("bootstrap.servers", kafkas) ;
                props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer") ;
                props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer") ;
                props.setProperty("group.id", "default") ;
                props.setProperty("auto.offset.reset", "earliest") ;
                props.setProperty("enable.auto.commit", "false") ;
                KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props) ;
                consumer.subscribe(Arrays.asList(topic)) ;
                while(true) {
                    for(ConsumerRecord<String, String> record: consumer.poll(Duration.ofMillis(100))) {
                        try {
                            Object key = record.key() ;
                            Object val = new ObjectMapper().readValue(record.value(), Object.class) ;

                            Map<String, Object> ruleValInfo = (Map<String, Object>) val;
                            String msgType = ruleValInfo.get("msgType").toString() ;
                            if (!msgType.equals("2")) {
                                Map<String, Object> exprInfo = new HashMap<String, Object>() ;
                                exprInfo.put("raw", val) ;
                                StateHandler handler = mkHandler(val) ;
                                exprInfo.put("handler", handler) ;

                                rules.put(key, exprInfo) ;
                            } else {
                                rules.remove(key) ;
                            }
                        } catch(Exception e) {
                            e.printStackTrace() ;
                            // System.exit(-1) ;
                        }
                    }
                }
            }
        }).start() ;
        Thread.sleep(1000 * 5) ;
    }
    static class PointConverter {
        private Schema schema = null ;
        public PointConverter(Schema schema) {
            this.schema = schema ;
        }
        private GenericContainer decodeAvro(boolean isJson, byte[] bs) throws Exception {
            return isJson? Avro.fromJson(schema, new String(bs)) : Avro.fromCompressedBin(schema, bs) ;
        }
        private List<Object> encodeJson(GenericContainer avro) throws Exception {
            List<Object> result = new ArrayList<Object>() ;

            Map<String, Object> newJson = new HashMap<String, Object>() ;
            List<Map<String, Object>> json = new ObjectMapper().readValue(Avro.toJson(schema, avro), List.class) ;
            info("json:" + json) ;
            String platformRaw = json.get(0).keySet().iterator().next() ;
            String platform = extractPlatformName(platformRaw.split("\\.")[0]) ;
            Map<String, Object> body = (Map<String, Object>)json.get(0).values().iterator().next() ;
            Object common = body.get("common") ;
            List<Map<String, Object>> events = (List<Map<String, Object>>) body.get("events") ; 

            for(Map<String, Object> eventUnion :  (List<Map<String, Object>>) body.get("events")) {
                Object event = eventUnion.values().iterator().next() ;
                newJson.put("platform", platform) ;
                newJson.put("common", common) ;
                newJson.put("event", event) ;
                result.add(newJson) ;
            }
            return result ;
        }
    }
    static class PointWriter {
        private KafkaProducer<byte[], byte[]> kafka = null ;
        private String topic = null ;
        public PointWriter(String kafkas, String topic) {
            info("kafkas:" + kafkas + " -> topic: " + topic) ;
            Properties props = new Properties() ;
            props.put("bootstrap.servers", kafkas);
            props.put("key.serializer",   "org.apache.kafka.common.serialization.ByteArraySerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
            this.kafka = new KafkaProducer<byte[], byte[]>(props) ;
            this.topic = topic ;
        }
        public void writeChannel(byte[] key, byte[] val) {
            info("write topic:" + topic) ;
            kafka.send(new ProducerRecord<byte [], byte []>(topic, key, val));
        }
    }
    static class Utils {
        private static void _flattenJson(String parent, Object json, Map<String, Object> collector) {
            if (json instanceof Map) {
                for(Map.Entry<Object, Object> entry: ((Map<Object, Object>) json).entrySet()) {
                    String key = entry.getKey().toString() ;
                    String nextParent = "__delay__" ;
                    if (key.equals("string") || key.equals("long") || key.equals("double")) {
                        nextParent = parent ;
                    } else {
                        nextParent = key ;
                    }
                    _flattenJson(nextParent, entry.getValue(), collector) ;
                }
            } else if (parent == null) {
            } else {
                collector.put(parent, json) ;
            }
        }
        public static Map<String, Object> flattenJson(Object json) {
            Map<String, Object> collector = new HashMap<String, Object>() ;
            _flattenJson(null, json, collector) ;
            return collector ;
        }
    }
    static class OnlineHandler implements Runnable {
        private BlockingQueue<OnlineMessage> channel = null ;
        private App app = null ;
        private PointConverter converter = null ;
        private PointWriter writer = null ;

        public OnlineHandler(BlockingQueue<OnlineMessage> channel, App app, PointConverter converter, PointWriter writer) {
            this.channel = channel ;
            this.app = app ;
            this.converter = converter ;
            this.writer = writer ;
        }

        public Object doRuleEngine(Object json, Object ruleKey, Object ruleBody) throws Exception {
            info("ruleKey:" + ruleKey) ;

            Map<String, String> result = new HashMap<String, String> () ;

            Map<String, Object> body = Utils.flattenJson(json) ;


            result.put("platform", body.get("platform").toString()) ;
            result.put("Apex_id", body.get("apex_id").toString()) ;

            Map<String, Object> ruleBodyInfo = (Map<String, Object>) ruleBody ;
            Map<String, Object> ruleValInfo = (Map<String, Object>) ruleBodyInfo.get("raw") ;

            String msgType = ruleValInfo.get("msgType").toString() ;
            if (msgType.equals("2")) return null ;

            StateHandler handler = (StateHandler) ruleBodyInfo.get("handler") ;

            boolean is_true = handler.refresh(body) ;
            if (!is_true) return null ;

            for(String bizAttr: (List<String>) ruleValInfo.get("bizAttribute")) {
                 String fVal = body.get(bizAttr) == null ? null : body.get(bizAttr).toString() ;
                 result.put(bizAttr, fVal);
            }
            return result ;
        }
        public void writeChannel(Object json, Object engineResult, Object ruleKey, Object ruleBody) throws Exception {
            Map<String, String> eventData = new HashMap<String, String>() ;

            Map<String, String> engineResultInfo = (Map<String, String>) engineResult ;
            eventData.putAll(engineResultInfo) ;

            Map<String, Object> ruleBodyInfo = (Map<String, Object>) ruleBody ;
            Map<String, Object> ruleValInfo = (Map<String, Object>) ruleBodyInfo.get("raw") ;
            Map<String, String> payload = (Map<String, String>) ruleValInfo.get("extParam") ;
            eventData.putAll(payload) ;

            eventData.put("subscription_event_code", ruleValInfo.get("complexEventCode").toString()) ;
            eventData.put("message_type", "2") ;

            String apexId = engineResultInfo.get("Apex_id").toString() ;

            info("[debug] ======== write channel ========") ;
            info("[debug] apex_id:" + apexId) ;
            info("[debug] eventData:" + eventData) ;

            writer.writeChannel(apexId.getBytes(), new ObjectMapper().writeValueAsBytes(eventData));
        }

        public void run() {
            while(true) {
                try {
                    OnlineMessage message = this.channel.take() ;
                    info("receive one message...") ;
                    GenericContainer avro = converter.decodeAvro(message.isJson(), message.getContent()) ;
                    info("avro:" + avro) ;
                    for(Object json: converter.encodeJson(avro)) {
                        for(Map.Entry<Object, Object> rule: app.rules.entrySet()) {
                            info("rule:" + rule) ;
                            Object engineResult = doRuleEngine(json, rule.getKey(), rule.getValue()) ;
                            info("engineResult:" + engineResult) ;
                            if(engineResult != null) {
                                writeChannel(json, engineResult, rule.getKey(), rule.getValue()) ;
                            }
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace() ;
                }
            }
        }

    }

    static class OnlineService extends OnlineGrpc.OnlineImplBase {
        private BlockingQueue<OnlineMessage> channel = null ;
        public OnlineService(BlockingQueue<OnlineMessage> channel) {
            this.channel = channel ;
        }
        @Override
        public void createAvroBulk(AvroBulkCreate create, StreamObserver<DefaultReply> ob)  {
            try {
                OnlineMessage message = new OnlineMessage(false, create.getBody().toByteArray()) ;
                channel.add(message) ;
                responseGrpc(ob, 0, "accept") ;
            } catch (Exception e) {
                e.printStackTrace();
                responseGrpc(ob, 1, String.format("error_info: %s", e)) ;
            }
        }
        @Override
        public void createJsonBulk(JsonBulkCreate create, StreamObserver<DefaultReply> ob) {
            try {
                OnlineMessage message = new OnlineMessage(true, create.getBody().toByteArray()) ;
                channel.add(message) ;
                responseGrpc(ob, 0, "accept") ;
            } catch (Exception e) {
                e.printStackTrace();
                responseGrpc(ob, 1, e.getMessage()) ;
            }
        }
        public void responseGrpc(StreamObserver<DefaultReply> ob, int code, String message) {
            DefaultReply reply = DefaultReply.newBuilder()
                .setCode(code)
                .setMessage(message)
                // .setDetails("")
                // .setOrigin("")
                .build();
            ob.onNext(reply);
            ob.onCompleted();
        }
    }
    static class OnlineMessage {
        private boolean isJson = false ;
        private byte[] content = null ;
        public OnlineMessage(boolean isJson, byte[] content) {
            this.isJson = isJson ;
            this.content = content ;
        }
        public boolean isJson() {
            return isJson ;
        }
        public byte[] getContent() {
            return content ;
        }
    }

    public static void main(String[] args) throws Exception {
        Dotenv dotenv = Dotenv.configure().directory("meta-lib-static/env/.env").load();
        String online_nodes = Common.resolveEnv(dotenv, "ONLINE_SERVER_HOST", "127.0.0.1");
        String myIp = Common.resolveMyIp(online_nodes);
        if (myIp == null) {
            error("ONLINE_SERVER_HOSTS not match! skip");
            System.exit(0);
        }
  
        String portStr = Common.resolveEnv(dotenv, "ONLINE_SERVER_PORT", null);
        if (portStr == null) {
            error("ONLINE_SERVER_PORT not exist!");
            System.exit(1);
        }
        String kafkas = Common.resolveEnv(dotenv, "ONLINE_SERVER_KAFKAS", "127.0.0.1:9092") ;
        String ruleTopic = Common.resolveEnv(dotenv, "ONLINE_SERVER_KAFKA_RULE_TOPIC", "data_channel_dev_me_event_rule") ;
        info("kafkas:" + kafkas) ;

        String channelKafkas = Common.resolveEnv(dotenv, "ONLINE_SERVER_CHANNEL_KAFKAS", "127.0.0.1:9092") ;
        String dataTopic = Common.resolveEnv(dotenv, "ONLINE_SERVER_CHANNEL_KAFKA_DATA_TOPIC", "data_channel_dev_me_event_data") ;
        String queueSizeText = Common.resolveEnv(dotenv, "ONLINE_SERVER_CHANNEL_QUEUE_SIZE", "100000") ;
        String handlerThreadNumText = Common.resolveEnv(dotenv, "ONLINE_SERVER_HANDLER_THREAD_NUM", "100") ;

        int port = Integer.parseInt(portStr) ;
        int queueSize = Integer.parseInt(queueSizeText) ;
        int handlerThreadNum = Integer.parseInt(handlerThreadNumText) ;

        App app = new App() ;
        app.asyncLoadRules(kafkas, ruleTopic) ;

        Schema schema = Avro.readAvscSchema("meta-lib-static/avsc/dcSdkPlatformBulk.avsc") ;
        PointConverter converter = new PointConverter(schema) ;
        PointWriter writer = new PointWriter(channelKafkas, dataTopic) ;

        BlockingQueue<OnlineMessage> channel = new LinkedBlockingQueue<OnlineMessage>(queueSize) ;
        
        for(int i = 0; i < handlerThreadNum; i++) {
            Thread onlineTask = new Thread(new OnlineHandler(channel, app, converter, writer)) ;
            onlineTask.start() ;
        }

        info(String.format("start server at %s", port));
        Server.builder()
            .http(port)
            .service("/online/health", (ctx, req) -> HttpResponse.of(HttpStatus.OK))
            .service(
              GrpcService.builder()
              .addService(new OnlineService(channel))
              .build()
            )
            .build().start() ;
            

/*
        Server server = ServerBuilder.forPort(port).addService(new OnlineService(channel)).build();
        info(String.format("start server at %s", port));

        server.start();
        server.awaitTermination();
*/
        
    }
}
